package org.geektime.reactive.message.streams.v1;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * @author lanxing
 */
public class DefaultSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;

    private int count = 0;
    @Override
    public void onSubscribe(Subscription s) {
        subscription = s;
    }

    @Override
    public void onNext(T t) {
        count ++;
        if(count > 3 ){
            subscription.cancel();
            System.out.println("本次数据发布已忽略:"+ t);
            return;
        }
        System.out.println("收到数据："+ t);
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("遇到异常:"+ t);
    }

    @Override
    public void onComplete() {
        System.out.println("接收数据完成");

    }
}

